
package node;

import java.rmi.AccessException;
import java.rmi.AlreadyBoundException;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Scanner;

import duplicatedataStorage.DuplicateStores;


import log.Logger;
import messaging.InterNodeMessage;
import messaging.MessageChannels;
import messaging.MessageHashing;
import messaging.TokenNodeMessage;
import messaging.TokenPacket;
import messaging.TokenType;
import rebalance.Rebalance;
import remote.RemoteDistributedHashMap;
import remote.UnicastRemoteDistributedHashMap;

/**
 * This class is the manager for all nodes virtual nodes that this actual node simulates.
 * It stores all the information of all the nodes that it manages.
 * It is the launching point for the start of a ring as well as NodeAdditions.
 * 
 * @author Team 4
 * @version Spring 2013
 * 
 */
public final class NodeManager
{
  /**
   * Node id/number of the actual node for this set of nodes managed by the
   * NodeManager.
   */
  public int actual_node_number;
  /**
   * The predecessor node to send data to.
   */
  public int predecessor_actual_node;
  /**
   * An instance of MultipleStores.
   */
  public DuplicateStores duplicatestores;
  /**
   * Stores the wait time (defaults to 20 seconds).
   */
  private long DEFAULT_SLEEP_TIME = 20000;
  /**
   * Host name of rmi registry.
   */
  public String HOST;
  /**
   * port number of registry.
   */
  public int PORT;
  /**
   * Remote registry name
   */
  public String REMOTE;
  /**
   * Stores the name of the logger file for this node.
   */
  public String LOGGER_FILE_NAME;
  /**
   * The maximum acceptable port number.
   */
  private static final int MAX_PORT = 65535;

  /**
   * The minimum acceptable port number.
   */
  private static final int MIN_PORT = 1025;
  /**
   * The hash map that stores all the nodes and the array list of its
   * neighbours.
   */
  public HashMap<Integer, ArrayList<Integer>> node_connections =
      new HashMap<Integer, ArrayList<Integer>>();
  /**
   * Stores a reference to Message Hashing.
   */
  public MessageHashing hashhelp;
  /**
   * The array list of all nodes this manager manages.
   */
  public ArrayList<Integer> node_set = new ArrayList<Integer>();
  /**
   * Hash map that stores nodes and its corresponding receiving channel class.
   */
  public HashMap<Integer, MessageChannels> node_channel =
      new HashMap<Integer, MessageChannels>();
  /**
   * The hash map that stores all the nodes this node manager manages.
   */
  public Map<Integer, Boolean> real_virtual_node_info =
      new HashMap<Integer, Boolean>();
  /**
   * A hash map that stores all the implemented remote interfaces.
   */
  public Map<Integer, UnicastRemoteDistributedHashMap> node_stubs =
      new HashMap<Integer, UnicastRemoteDistributedHashMap>();
  /**
   * An array of remote interfaces.
   */
  public UnicastRemoteDistributedHashMap[] remoteArray;
  /**
   * Registry object.
   */
  public static Registry registry;
  /**
   * Stores the registry name.
   */
  public static String REGISTRY_NAME = "RemoteRegistry";
  /**
   * The total number of nodes in the chord.
   */
  public static int total_nodes_chord;
  /**
   * Boolean to indicate whether this actual node is in re-balancing.
   */
  public boolean rebalancing = false;
  /**
   * This method sets the re-balancing part to true if NodeManager begins re-balancing.
   */
  public void setrelanacing(boolean rebalance)
  {
    this.rebalancing = rebalance;
  }
  /**
   * Re-balance.
   */
  public synchronized void rebalance()
  {
    if (!this.rebalancing)
    {
      //check if i need to really rebalance.
      try
      {
        Logger.out(this.actual_node_number + " checking before rebalancing");
        int dest_node = this.duplicatestores.successor_node_set.get(0);
        int dest_to_send = this.hashhelp.optimalNodeFind(this.actual_node_number, dest_node);
        final String dest_node_str = this.rmi_NodeFind(dest_to_send);
        InterNodeMessage message = 
            new InterNodeMessage(this.actual_node_number, dest_node,this.actual_node_number, dest_to_send,20);
        final RemoteDistributedHashMap to_send =
            (RemoteDistributedHashMap) NodeManager.registry.lookup(dest_node_str);
          to_send.send(message);
      }
      catch (RemoteException e)
      {
        Logger.out(this.actual_node_number + " Received rebalance message ");
        this.rebalancing = true;
        final int old_nodeset_size = node_set.size();
        final Rebalance reb = new Rebalance(this);
        reb.rebalance();
        messageChannelSetup(old_nodeset_size);
        QueueThreadStart(old_nodeset_size);
        reb.sendRebalanceUpdatesPredecessor();
        this.duplicatestores.initializeStorage();
        reb.sendRebalanceUpdatesSuccessor(); 
        //reb.registryupdates(old_nodeset_size);
        
        this.setrelanacing(false);
      }
      catch (NotBoundException e)
      {
        // TODO Auto-generated catch block
        e.printStackTrace();
      }
      catch (InterruptedException e)
      {
        // TODO Auto-generated catch block
        e.printStackTrace();
      }
    }
    else
    {
      Logger.out(this.actual_node_number + 
                 " Actual Node manager already in the process of rebalancing ");
    }
  }
  /**
   * Setting the predecessor for this node set.
   */
  public void setpredecessor()
  {
    duplicatestores = new DuplicateStores(this,this.actual_node_number);
    this.predecessor_actual_node = duplicatestores.findPredecessor(this.actual_node_number);
  }
  /**
   * This method creates virtual nodes if needed.
   * 
   * @throws RemoteException
   * @throws AlreadyBoundException
   * @throws NotBoundException
   */
  private void createVirtualNodes() throws RemoteException, NotBoundException,
      AlreadyBoundException
  {
    int count = 1;
    int i = actual_node_number + 1;
    if (i == total_nodes_chord)
    {
      i = 0;
    }
    while (registry.lookup(Integer.toString(i) + "A") == null)
    {
      remoteArray[count] = new UnicastRemoteDistributedHashMap(i, false, actual_node_number,this);
      node_stubs.put(i, remoteArray[count]);
      real_virtual_node_info.put(i, false);
      final RemoteDistributedHashMap stub =
          (RemoteDistributedHashMap) UnicastRemoteObject.exportObject(remoteArray[count], 0);
      count++;
      String rmi_node_name = new Integer(i).toString() + "V";
      registry.bind(rmi_node_name, stub);
      Logger.writeContent(LOGGER_FILE_NAME, rmi_node_name +
                           " (virtual) has been added to the registry");
      Logger.out(rmi_node_name + " (virtual) is added to the registry");
      node_set.add(i);
      i = i + 1;
      if (i == total_nodes_chord)
      {
        i = 0;
      }
      if (i == actual_node_number)
      {
        break;
      }
    }
  }
  /**
   * This method actually creates the stub for the actual node and binds it to
   * the registry.
   * 
   * @param the_node_num the number of current node
   * @param the_host the host name
   * @param the_port the port number
   * @param the_registry_name the name of registry
   */
  public void remoteObjectCreation(final int the_node_num, final String the_host,
                                           final int the_port, final String the_registry_name)
  {
    int i = 0;
    remoteArray = new UnicastRemoteDistributedHashMap[total_nodes_chord];
    try
    {
      remoteArray[i] =
          new UnicastRemoteDistributedHashMap(the_node_num, true, actual_node_number, this);
      node_stubs.put(the_node_num, remoteArray[i]);
      final RemoteDistributedHashMap stub =
          (RemoteDistributedHashMap) UnicastRemoteObject.exportObject(remoteArray[i], 0);
      final String rmi_node_name = new Integer(the_node_num).toString() + "A";
      registry.bind(rmi_node_name, stub);
      Logger.writeContent(LOGGER_FILE_NAME, rmi_node_name +
                          " (actual) added to the registry.");
      Logger.out(rmi_node_name + " (actual) added to the registry");
      node_set.add(the_node_num);
      if (actual_node_number == the_node_num)
      {
        real_virtual_node_info.put(the_node_num, true);
      }
      else
      {
        real_virtual_node_info.put(the_node_num, false);
      }
    }
    catch (final AccessException e)
    {
      Logger.writeContent(LOGGER_FILE_NAME, "issue accessing registry:" + e);
      Logger.out("issue accessing registry: " + e);
      System.exit(1);
    }
    catch (final AlreadyBoundException e)
    {
      Logger.writeContent(LOGGER_FILE_NAME, "already bound??: " + e);
      Logger.out("already bound???: " + e);
      System.exit(1);
    }
    catch (final RemoteException e)
    {
      Logger.writeContent(LOGGER_FILE_NAME, "a problem occurred, because: " + e);
      Logger.out("a problem occurred: " + e);
      System.exit(1);
    }
  }
  /**
   * This constructors exports the stub to rmi registry and binds it with the
   * given node number.
   * 
   * @param the_host stores the host name
   * @param the_port stores the port number
   * @param the_registry_name stores the name of the RMI registry
   */
  public NodeManager(final String the_host, final int the_port, final String the_registry_name)
  {
    try
    {
      final Registry bootstrap = LocateRegistry.getRegistry(the_host, the_port);
      registry = (Registry) bootstrap.lookup(the_registry_name);
    }
    catch (final RemoteException e)
    {
      Logger.err("a problem occurred:" + e);
      System.exit(1);
    }
    catch (final NotBoundException e)
    {
      Logger.err("register name not bound:" + e);
      System.exit(1);
    }
  }
  /**
   * This method creates all the channels for all the nodes that this node
   * manager manages.
   * @param start_point - the node number from which to start message channel set up
   */
  public void messageChannelSetup(final int start_point) 
  {
    try
    {
      for (int i = start_point; i < node_set.size(); i++)
      {
        final int current_node_no = node_set.get(i);
        final MessageChannels channels = new MessageChannels(current_node_no, this);
        this.node_connections.put(current_node_no,
                                       channels.get_my_neighbours(current_node_no));
        channels.createChannels();
        node_channel.put(current_node_no, channels);
      }
      hashhelp = new MessageHashing(node_connections);
    }
    catch (final NotBoundException e)
    {
      Logger.writeContent(LOGGER_FILE_NAME, " the given name is not bound " + e);
      System.exit(1);
    }
    catch (final RemoteException e)
    {
      Logger.writeContent(LOGGER_FILE_NAME, " Remote exception occured " + e);
      System.exit(1);
    }
  }
  /**
   * This method will start the thread for queue processor to listen to messages
   * on each queue.And storing all the threads that this NodeManager manages in  a
   * hash map.
   * @param start_point - the node number from which to start spawning threads.
   * 
   */
  public void QueueThreadStart(final int start_point)
  {
    final Thread[] t = new Thread[node_set.size()];
    final Runnable[] queue = new QueueProcessor[node_set.size()];
    for (int i = start_point; i < node_set.size(); i++)
    {
      queue[i] = new QueueProcessor(node_set.get(i),this);
      t[i] = new Thread(queue[i]);
      t[i].start();
    }
  }
  /**
   * Given a node number this method will find its rmi name in registry by appending
   * a V or a A to it corresponding to actual/virtual.
   * @return String - the registry name for this node number.
   */
  public String rmi_NodeFind(final int node_num) 
  {
    final String rmi_name_actual = new Integer(node_num).toString() + "A";
    final String rmi_name_virtual = new Integer(node_num).toString() + "V";
    try
    {
      if (NodeManager.registry.lookup(rmi_name_actual) == null)
      {
        return rmi_name_virtual;
      }
      else
      {
        return rmi_name_actual;
      }
    }
    catch (final RemoteException | NotBoundException e)
    {
     // Logger.writeContent(LOGGER_FILE_NAME, 
       //                   " Remote exception occured or the given name is not bound " + e);
      System.exit(1);
    }
    return rmi_name_virtual;
  }
    /**
   * The main() method. Accepts the input from the user and creates a actual node. It 
   * then waits for a predefined time period and then simulates the virtual
   * nodes.
   * This is also the launching point for Adding node. It will check the number of nodes in
   * the registry and will launch - 
   * either start a ring if there is not a full set of nodes in ring.
   * or launch the LaunchAddNode if there is a full set but we wish to add one more
   * actual node to the chord.
   * 
   * @param the_args The first argument is the value of m,the_args The registry host name,
   * the_args The registry port number in the registry host, 
   * the_args The optional remote registry name.
   * @throws InterruptedException 
   */
  public static void main(final String[] the_args) throws InterruptedException
  {
    try
    {
      String remote = REGISTRY_NAME;  
      if (the_args.length < 3)
      {
        System.out.println("Usage <m> <rmi registry host> <rmi registry port> " +
                          "<remote rmiregistry name>");
        System.exit(1);
      }
      if (the_args.length == 4)
      {
        remote = the_args[3];
        REGISTRY_NAME = remote;
      }
      String host = the_args[1];
      int port = Integer.parseInt(the_args[2]);
      if (port < MIN_PORT || MAX_PORT < port)
      {
        System.err.println("port must be between 1025 and 65535 inclusive");
        System.exit(1);
      }
      total_nodes_chord = (int) Math.pow(2, Integer.parseInt(the_args[0]));
      final Registry bootstrap = LocateRegistry.getRegistry(host, port);
      registry = (Registry) bootstrap.lookup(remote);
      final Scanner scanner = new Scanner(System.in);
      if (registry.list().length == NodeManager.total_nodes_chord)
      {
        System.out.println(" List of nodes in registry are - " + 
            Arrays.toString(registry.list()));
        System.out.println(" the chord ring is full. Do you want to add/remove/token nodes? ");
        System.out.println(" Type add to add actual node. Type remove to remove actual node.");
        final String todo = scanner.nextLine();
        if ("ADD".equalsIgnoreCase(todo))
        {
          final LaunchAddNode launch = new LaunchAddNode(host, port, remote);
          launch.Adding_Node();
          System.exit(0);
        }
        if ("REMOVE".equalsIgnoreCase(todo))
        {
          LaunchRemoveNode remove_node = new LaunchRemoveNode();
          int node_num = remove_node.Remove_Node();
          System.out.println("NodeRemoval : Node Removal started");
          System.out.println("NodeRemoval : Node no" + node_num);
          final RemoteDistributedHashMap to_send =
              (RemoteDistributedHashMap) NodeManager.registry.lookup(node_num+"A");
          System.out.println("NodeRemoval : Node Removal Sending"); 
          to_send.removeNode();
          System.exit(0);   
        }
        
      }
      else
      {
        Logger.out("Located registry.");
        System.out.println(" List of nodes in registry are - " + 
          Arrays.toString(registry.list()));
        System.out.println(" Enter the number of actual node to create.");
        final NodeManager nm = new NodeManager(host, port, remote); 
        final String node_num = scanner.nextLine();
        nm.actual_node_number = Integer.parseInt(node_num);
        nm.LOGGER_FILE_NAME = "LOG-" + nm.actual_node_number + ".txt";
        if (nm.actual_node_number < 0 || total_nodes_chord <= nm.actual_node_number)
        {
          System.err.println(" The node number must be between 0 and " +
            (total_nodes_chord - 1) + "inclusive");
          System.exit(1);
        }
        Logger.writeContent(nm.LOGGER_FILE_NAME, "the total nodes in the current chord is " +
                                            total_nodes_chord);
        Logger.out("the total nodes in the chord is " + total_nodes_chord);
        nm.remoteObjectCreation(nm.actual_node_number, host, port, remote);
        try
        {
          Thread.sleep(nm.DEFAULT_SLEEP_TIME);
        }
        catch (final InterruptedException e)
        {
          Logger.writeContent(nm.LOGGER_FILE_NAME, 
                              " Interrupted when waiting for nodes to bind  " + e);
        }
        nm.createVirtualNodes();
        nm.messageChannelSetup(0);
        nm.QueueThreadStart(0);
        nm.setpredecessor();
        Logger.out("Ready for rmi calls from client ");
        Logger.writeContent(nm.LOGGER_FILE_NAME, " Ready for rmi calls from client");
      }
    }
    catch (final AlreadyBoundException e)
    {
      Logger.out("already bound???: " + e);
      System.exit(1);
    }
    catch (final RemoteException e)
    {
      Logger.err("a problem occurred:" + e);
      e.printStackTrace();
      System.exit(1);
    }
    catch (final NotBoundException e)
    {
      Logger.err("register name not bound:" + e);
      System.exit(1);
    }
  }
}